package mqConsumer

import (
	"gin_easy_starter/internal/mq"
	"github.com/IBM/sarama"
	"log"
)

func ConsumeMessages(consumer sarama.Consumer, topicName string, partitionNum int32, handleReceive func(message *sarama.ConsumerMessage), handleError func()) {
	// Specify your Kafka topic
	topic := topicName
	err := mq.CreateKafkaTopic(topic, 0, 1)
	if err != nil {
		log.Fatal("Error creating topic ", err)
	}
	// Create a partition consumer for the specified topic and partition
	partitionConsumer, err := consumer.ConsumePartition(topic, partitionNum, sarama.OffsetOldest)
	if err != nil {
		log.Fatal("Error creating partition consumer:", err)
	}

	defer func() {
		if err := partitionConsumer.Close(); err != nil {
			log.Fatal("Error closing partition consumer:", err)
		}
	}()

	log.Printf("Kafka consumer started. listen on: %s ,patition : %d", topic, partitionNum)

	// Start consuming messages from Kafka
	for {
		select {
		case msg := <-partitionConsumer.Messages():
			// Handle the received message
			handleReceive(msg)
			log.Printf("Received message: %s\n", msg.Value)
		case err := <-partitionConsumer.Errors():
			// Handle any errors that occur during consumption
			handleError()
			log.Println("Error:", err)
		}
	}
}
